package cn.itcast.com.java.wc;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;

public class JavaWordCount {
    public static void main(String[] args) {
        //初始化对象
        SparkConf sparkConf = new SparkConf().setAppName("JavaWordCount").setMaster("local[2]");
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);
        //设置日志级别
        jsc.setLogLevel("WARN");

        //加载文件
        JavaRDD<String> data = jsc.textFile("file:///F:\\text.txt");

        //切分并压平
        JavaRDD<String> words = data.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String line) throws Exception {
                String[] words = line.split(" ");
                return Arrays.asList(words).iterator();
            }
        });
        //单词组成元组 (单词，1)
        JavaPairRDD<String, Integer> wordAnd1 = words.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String word) throws Exception {
                return new Tuple2<>(word, 1);
            }
        });
        //统计汇总 不管key  value进行累加
        JavaPairRDD<String, Integer> wc = wordAnd1.reduceByKey(new Function2<Integer, Integer, Integer>() {
            //将单词出现的次数进行累加
            @Override
            public Integer call(Integer integer, Integer integer2) throws Exception {
                return (integer+integer2);
            }
        });

        //将统计出来的次数进行排序[(love,3), (too,1), (i,3), (u,3)]

        //打印结果
        System.out.println(wc.collect());
        //释放资源
        jsc.close();
    }
}
